package com.test.rxjavademo;


import org.jetbrains.annotations.NotNull;

import java.util.ArrayList;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;

class CreateOperatorDemo {

    private static Observer observer = new Observer<Object>() {//Observer观察者，订阅subscribe了被观察者Observable

        @Override
        public void onSubscribe(@NotNull Disposable disposable) {
            System.out.println("1、onSubscribe,首次订阅时会回调:" + disposable);
        }

        @Override
        public void onNext(@NotNull Object o) {
            System.out.println("onNext:" + o);
        }

        @Override
        public void onError(@NotNull Throwable throwable) {
            System.out.println("onError:" + throwable);
        }

        @Override
        public void onComplete() {
            System.out.println("onComplete");
        }
    };


    public static void createTest(){
        System.out.println("createTest");

        //Observable被观察者
        Observable.create(new ObservableOnSubscribe<Object>() {
            @Override
            public void subscribe(@NotNull ObservableEmitter<Object> e) throws Exception {
                System.out.println("2、subscribe,有观察者订阅时会回调:" + e); //被观察者的操作
                e.onNext(1);
                e.onNext(2);
                e.onNext("str333");
                e.onError(new Throwable("主动抛异常"));//onError  和 onComplete是互斥的
                e.onComplete();//完成后，就不能再发射了
                e.onNext("4444");

                /**用在网络封装**/
            }
        }).subscribe(new Observer<Object>() {//Observer观察者，订阅subscribe了被观察者Observable

            @Override
            public void onSubscribe(@NotNull Disposable disposable) {
                System.out.println("1、onSubscribe,首次订阅时会回调:" + disposable);
            }

            @Override
            public void onNext(@NotNull Object o) {
                System.out.println("onNext:" + o);
            }

            @Override
            public void onError(@NotNull Throwable throwable) {
                System.out.println("onError:" + throwable);
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });
    }



    public static void consumerTest(){
        System.out.println("createTest");
        //Observable被观察者
        Disposable disposable = Observable.create(new ObservableOnSubscribe<Object>() {
            @Override
            public void subscribe(@NotNull ObservableEmitter<Object> e) throws Exception {
                System.out.println("2、subscribe,有观察者订阅时会回调:" + e) ;
                e.onNext(1);
                e.onNext(2);
                e.onNext("str333");
                e.onError(new Throwable("主动抛异常"));//onError  和 onComplete是互斥的
                e.onComplete();//完成后，就不能再发射了
                e.onNext("4444");
            }
        }).subscribe(new Consumer<Object>() {

            @Override
            public void accept(@NotNull Object o) throws Exception {
                System.out.println(" consumer accept Object:" + o);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(@NotNull Throwable throwable) throws Exception {
                System.out.println(" consumer accept throwable:" + throwable);
            }
        });
    }

    public static void justTest(){//just
        Observable.just("A","b",11)//just被观察者创建符号
                .subscribe(observer);
    }


    public static void fromArrayTest(){//fromArray
        Observable.fromArray("A","b",11)//just被观察者创建符号
                .subscribe(observer);
    }

    public static void fromIterableTest(){//fromIterable
        ArrayList<Object> list = new ArrayList<>();
        list.add("a");
        list.add(1);
        list.add(3);
        Observable.fromIterable(list)//just被观察者创建符号
                .subscribe(observer);
    }


    public static void fromFutureTest(){//fromFuture
        System.out.println(" fromFutureTest:");
        Observable.fromFuture(new Future<Object>() {
            @Override
            public boolean cancel(boolean b) {
                System.out.println(" cancel:" + b);
                return false;
            }

            @Override
            public boolean isCancelled() {
                System.out.println(" isCancelled:");
                return false;
            }

            @Override
            public boolean isDone() {
                System.out.println(" isCancelled" );
                return false;
            }

            @Override
            public Object get() throws ExecutionException, InterruptedException {
                System.out.println(" get thread:" + Thread.currentThread().getName());
                return null;
            }

            @Override
            public Object get(long l, TimeUnit timeUnit) throws ExecutionException, InterruptedException, TimeoutException {
                System.out.println(" get" );
                return null;
            }
        }).subscribe(observer);
    }



    public static void fromCallable(){//fromCallable
        ArrayList<Object> list = new ArrayList<>();
        list.add("a");
        list.add(1);
        list.add(3);
        Observable.fromCallable(new Callable<Object>() {

            @Override
            public Object call() throws Exception {
                return "fromCallable test ";
            }
        })//just被观察者创建符号
                .subscribe(observer);
    }


    public static void main(String[] args) {
//        createTest();
//        consumerTest();
//        justTest();
//        fromArrayTest();
//        fromIterableTest();
//        fromFutureTest();
        fromCallable();
    }
}
